package hit.edu.zjc.Operator.ValueFill;

import hit.edu.zjc.Tool.MessyCodeCheck;
import hit.edu.zjc.Tool.ObjectConvert;
import hit.edu.zjc.UserInterface.UserDef;
import hit.edu.zjc.marshalling.ByteSerializerDeserializer;

import java.io.ByteArrayInputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.Array;
import java.nio.ByteBuffer;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Random;

import edu.uci.ics.hyracks.api.comm.IFrameWriter;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.job.JobSpecification;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.Integer64SerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputOperatorNodePushable;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputUnaryOutputOperatorNodePushable;
import edu.uci.ics.hyracks.dataflow.std.file.DelimitedDataTupleParserFactory;

/*
 * 在原ValueInvalidDetectOperatorDescriptor的基础上，
 * 增加了对非EasyFill列的Gram表建立工作。
 * */
public class ValueInvalidDetectNullGramBuildOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {

    private static final long serialVersionUID = 1L;
    private ISerializerDeserializer[] DataISD_array=null;
    private UserDef[] UD_array=null;
    private Hashtable<String,HashSet<Integer>>[] GramMap=null;
    public ValueInvalidDetectNullGramBuildOperatorDescriptor(JobSpecification spec,ISerializerDeserializer[] isd_array,UserDef[] ud_array)throws HyracksDataException {
        super(spec, 1, 2);//0输出文本，1输出GramMap
        DataISD_array=isd_array;
        UD_array=ud_array;
        recordDescriptors[0] = new RecordDescriptor(DataISD_array);//输出文本
        recordDescriptors[1] = new RecordDescriptor(new ISerializerDeserializer[]{ByteSerializerDeserializer.INSTANCE});//输出GamMap
        if(DataISD_array.length!=UD_array.length)
        	 throw new HyracksDataException("UserDef[]:"+UD_array.length+"   DataISD[]:"+DataISD_array.length);
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * edu.uci.ics.hyracks.api.dataflow.IActivity#createPushRuntime(edu.uci.ics.hyracks.api.context.IHyracksTaskContext,
     * edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider, int, int)
     */
    @Override
    public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
            IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) throws HyracksDataException {

        final FrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), new RecordDescriptor(DataISD_array));

        return new AbstractUnaryInputOperatorNodePushable() {

            private ArrayTupleBuilder tupleBuilder;


            private FrameTupleAppender outputAppender;
            
            private final IFrameWriter[] writers = new IFrameWriter[outputArity];
      	  
            private ByteBuffer[] outputBuffer=new ByteBuffer[outputArity];

            @Override
            public void open() throws HyracksDataException {
                tupleBuilder = new ArrayTupleBuilder(DataISD_array.length);
                outputBuffer[0] = ctx.allocateFrame();
                outputBuffer[1] = ctx.allocateFrame();
                outputAppender = new FrameTupleAppender(ctx.getFrameSize());
                outputAppender.reset(outputBuffer[0], true);
                GramMap=(Hashtable<String, HashSet<Integer>>[])Array.newInstance(Hashtable.class, UD_array.length);
                for(int i=0;i<UD_array.length;i++)
                	{
                	GramMap[i]=new Hashtable<String,HashSet<Integer>>();
                	}
                for(IFrameWriter w:writers)
                	w.open();
            }

            @Override
            public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
                accessor.reset(buffer);
                int tupleCount = accessor.getTupleCount();
                int fieldCount = accessor.getFieldCount();
                int fieldSlotsLength=accessor.getFieldSlotsLength();
                byte[] byte_array=accessor.getBuffer().array();
                for (int i = 0; i < tupleCount; i++) {
                    int tupleStartOffset = accessor.getTupleStartOffset(i);//获得第i个Tuple的开始位置。
                    tupleBuilder.reset();
                    	 //对每个tuple中的每个field进行：UserDef的检查。
                    try {
                    	 int Data_ID=0;
	                    for(int fIdx=0;fIdx<fieldCount;fIdx++)
	                    	 {
	                    	int fieldstartoffset=accessor.getFieldStartOffset(i, fIdx);
	                    	int fieldendoffset=accessor.getFieldEndOffset(i, fIdx);
	                    	int fieldlength=accessor.getFieldLength(i, fIdx);
	                    	//String data_str= new String(byte_array,tupleStartOffset+fieldSlotsLength+fieldstartoffset,fieldlength);//将field从二进制中提取出来。
								String data_str=UTF8StringSerializerDeserializer.INSTANCE.deserialize( new DataInputStream(new ByteArrayInputStream(byte_array,tupleStartOffset+fieldSlotsLength+fieldstartoffset,fieldlength)));
	                    	//System.out.print("length:"+fieldlength+" "+data_str+",");
								if(fIdx==0)
								{
									Data_ID=Integer.parseInt(data_str);
								}
								boolean isMessy=MessyCodeCheck.isMessyCode(data_str);//检查乱码
	                    	if(isMessy)//乱码
	                    		{
	                    		data_str=null;//按空处理。
	                    		}
	                    	if(UD_array[fIdx].isCheck)
								{
	                    		
		                    	boolean isvalid=UD_array[fIdx].checkdata(data_str);//进行异常值检查。
		                    		//此处应当记录清洗日志。
		                    	if(!isvalid)//检测出异常，进行异常值处理,记录清洗日志
		                    		{
	                    			if(UD_array[fIdx].isEasyFill)//简单填充
	                    				data_str=UD_array[fIdx].correctdata();
	                    			else//非简单填充，应该建立Gram
	                    				{
	                    				data_str=null;
	                    				int[]cid=UD_array[fIdx].RelevantColumID;
	                    				for(int k=0;k<cid.length;k++)
	                    					{
	                    					int ColumID=cid[k];
	                    					int startoffset=accessor.getFieldStartOffset(i, ColumID);
	            	                    	int endoffset=accessor.getFieldEndOffset(i, ColumID);
	            	                    	int length=accessor.getFieldLength(i, ColumID);
	            	                    	String colum_str=UTF8StringSerializerDeserializer.INSTANCE.deserialize( 
	            	                    			new DataInputStream(new ByteArrayInputStream(
	            	                    					byte_array,tupleStartOffset+fieldSlotsLength+startoffset,length)));
	                    					
	            	                    	
	            	                    	
	            	                    	    //字符串的忽略大小写
	            	                    	colum_str=colum_str.toUpperCase();
	            	                    	    //去掉字符串空格。
	            	                    	colum_str=colum_str.replaceAll(" ","");
	            	                    		//将字符串，每个一个字符都加入GramMap中。
	            	                    	for(int data_str_i=0;data_str_i<colum_str.length();data_str_i++)
	            	                    		{
	            	                    		String key=colum_str.substring(data_str_i, data_str_i+1);
	            	                    		HashSet<Integer> Set=GramMap[ColumID].get(key);
	            	                    		if(Set==null)//键不存在。
	            	                    			{
	            	                    			Set=new HashSet<Integer>();
	            	                    			}
	                                			Set.add(Data_ID);
	                                			GramMap[ColumID].put(key, Set);
	            	                    		}
	            	                    	   //将字符串，每个2-gram都加入GramMap中
	            	                    	for(int data_str_i=0;data_str_i<colum_str.length()-1;data_str_i++)
	            	                    		{
	            	                    		String key_2=colum_str.substring(data_str_i, data_str_i+2);
	            	                    		HashSet<Integer> Set=GramMap[ColumID].get(key_2);
	            	                    		if(Set==null)
	            	                    			{
	            	                    			Set=new HashSet<Integer>();
	            	                    			}
	            	                    		Set.add(Data_ID);
	            	                    		GramMap[ColumID].put(key_2, Set);
	            	                    		}    					
	                    					
	                    					}
	                    				}
		                    		}
								}
	                    	if(data_str==null)//处理一下空值
                    			data_str="";
	                    	tupleBuilder.addField(DataISD_array[fIdx], data_str);//将最终结果写进去。
	                    	 }
                    } catch (Exception e) {
								e.printStackTrace();//产生异常的原因是字符串编码不是UTF-8的。
								System.out.println("产生异常,原因是字符串编码不是UTF-8.");
						 }                  
                    outputAppender.reset(outputBuffer[0], false);
                    if (!outputAppender.append(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray(), 0,
                            tupleBuilder.getSize())) {
                        FrameUtils.flushFrame(outputBuffer[0], writers[0]);
                        outputAppender.reset(outputBuffer[0], true);
                        if (!outputAppender.append(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray(), 0,
                                tupleBuilder.getSize())) {
                            throw new HyracksDataException(
                                    "Failed to copy an record into a frame: the record size is too large.");
                        }
                    }
                }
            }

            @Override
            public void fail() throws HyracksDataException {
            }

            @Override
            public void close() throws HyracksDataException {
                if (outputAppender.getTupleCount() > 0) 
                    {
                    FrameUtils.flushFrame(outputBuffer[0], writers[0]);
                    }
                byte[] objbytes=ObjectConvert.ObjectToByteArray(GramMap);
                WriteByteBuffer(1,objbytes);
                outputAppender = null;
                outputBuffer[0]= null;
                outputBuffer[1]=null;
                for(IFrameWriter w:writers)
                	w.close();
              }
            
            
            private void WriteByteBuffer(int index,byte[] data) throws HyracksDataException
        	{
        		outputBuffer[index].clear();
        	   if(data.length<outputBuffer[index].capacity())
           		{
               outputBuffer[index].put(data);
               FrameUtils.flushFrame(outputBuffer[index], writers[index]);
           		}
           else//一个frame装不下
           		{
           		int byteremain=data.length;
           		int i=0;
           		for(i=0;byteremain>=outputBuffer[index].capacity();i++)
           			{
           			outputBuffer[index].put(data,i*outputBuffer[index].capacity(),outputBuffer[index].capacity());
               		FrameUtils.flushFrame(outputBuffer[index], writers[index]);
               		outputBuffer[index].clear();
               		byteremain=byteremain-outputBuffer[index].capacity();
           			}
           		if(byteremain>0)
           			{
           			outputBuffer[index].put(data,i*outputBuffer[index].capacity(),byteremain);
           			FrameUtils.flushFrame(outputBuffer[index], writers[index]);
           			}
           		}
        	}
            

			@Override
			public void setOutputFrameWriter(int index, IFrameWriter writer,
					RecordDescriptor rDesc)
			{
				 writers[index] = writer;
				
			}
        };
    }
}